package com.herenit.order_model.utils.kafka;

import com.herenit.order_model.enums.ExceptionEnums;
import com.herenit.order_model.exception.MyException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducerUtils {

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    @Autowired
    private KafkaProducer<String, Object> producer;

    private static String PRIOR = "topic_prior_";

    private static String NORMAL = "topic_normal_";

    /**
     * 发送消息进一般队列
     *
     * @param message
     * @param institutionCode
     */
    public void send(String message, String institutionCode) {
        ProducerRecord<String, Object> record;
        for (int i = 0; i < 10; i++) {
            // 新建消息 指定topic为 一般队列
            record = new ProducerRecord(NORMAL + institutionCode, message + i);
            try {
                // 生产者发送消息进Kafka队列
                producer.send(record);
            } catch (Exception e) {
                throw new MyException(ExceptionEnums.KAFKA_SEND_ERROR);
            }
        }
    }

    /**
     * 发送消息进优先队列
     *
     * @param message
     * @param institutionCode
     */
    public void sendPrior(String message, Long recordId, String institutionCode) {
        ProducerRecord<String, Object> record;
        // 新建消息 指定topic为 优先队列
        record = new ProducerRecord(PRIOR + institutionCode, recordId + "", message);
        try {
            // 生产者发送消息进Kafka队列
            producer.send(record);
        } catch (Exception e) {
            throw new MyException(ExceptionEnums.KAFKA_SEND_ERROR);
        }
    }
}
